#include "umSource.h"
#include "umSink.h"
#include "umlog.h"

#define TARGET "TSSOUTCE"
#define MAXCACHESIZE 2000

#define DEBUGLEVEL tssourceLevel
static int tssourceLevel ;

UM_VOID setTsSourceDebug(UM_S32 level)
{
	tssourceLevel = level;
}

typedef struct
{
	UM_U8 data[1316 * 10 + 8];
	UM_S32 datasize;
	struct umapi_list_head head;
}S_BLOCK;
typedef struct
{	
	UM_BOOL flag, stopflag;
	UM_S32 blockCount;
	UM_S32 cacheflag;
	UM_HANDLE mutexHandle,semhandle, taskHandle,fileHandle;
	S_BLOCK *pblock;
	struct umapi_list_head sinks;
	struct umapi_list_head datablocks;
	struct umapi_list_head freeblocks;
}S_BASICCONTEXT;


static UM_S32 sendDataByCache(UM_VOID *pSource, AVPacket *pkt)
{	
	S_BASICCONTEXT *pHttp = GETCONTEXT(S_BASICCONTEXT *, pSource);
	umapi_os_mutex_lock(pHttp->mutexHandle, -1);
	S_BLOCK *pBlock = UM_NULL;
	struct umapi_list_head *pos, *n;
	umapi_list_for_each_safe(pos, n, &pHttp->freeblocks)
	{
		pBlock = umapi_list_entry(pos,S_BLOCK, head);
		umapi_list_del(&pBlock->head);
		break;
	}
	umapi_os_mutex_unlock(pHttp->mutexHandle);
	if(pBlock)
	{
		umapi_memcpy(&pBlock->data[6], pkt->data, pkt->size);
		pBlock->datasize = pkt->size;
		umapi_os_mutex_lock(pHttp->mutexHandle, -1);
		umapi_list_add(&pBlock->head, &pHttp->datablocks);
		pHttp->blockCount++;
		umapi_os_mutex_unlock(pHttp->mutexHandle);
		LOGI("Sourcep[%p] Report have block :%d", pHttp, pHttp->blockCount);
	}
	else
	{
		LOGI("no free pBlock");
	}
	return UM_SUCCESS;
}
static UM_VOID sendDatas(UM_VOID *pSource, AVPacket *pkt)
{
	S_BASICCONTEXT *pHttp = GETCONTEXT(S_BASICCONTEXT *, pSource);
	umapi_os_mutex_unlock(pHttp->mutexHandle);
	if(pHttp->flag)
	{
		struct umapi_list_head *pos, *n;
		umapi_list_for_each_safe(pos, n, &pHttp->sinks)
		{
			S_SINK *sink = umapi_list_entry(pos,S_SINK, head);
			if(sink->writePacket)
			{
				LOGD("send to sink ");
				sink->writePacket(sink, pkt);
			}
		}
	}
	umapi_os_mutex_unlock(pHttp->mutexHandle);
	return ;
}
static UM_S32 sendDataToSinks(UM_VOID *pSource, AVPacket *pkt)
{
	S_BASICCONTEXT *pHttp = GETCONTEXT(S_BASICCONTEXT *, pSource);
	if(pHttp->cacheflag)
		sendDataByCache(pSource, pkt);
	else
		sendDatas(pSource, pkt);
}

static UM_VOID handoutTask(UM_VOID *param)
{
	S_BASICCONTEXT *pHttp = (S_BASICCONTEXT*)param;
	S_BLOCK * pBlock = UM_NULL;
	while(pHttp->flag)
	{
		while(1)
		{
			pBlock = UM_NULL;
			if(!pHttp->flag)
				goto breakLoop;
			umapi_os_mutex_lock(pHttp->mutexHandle, -1);
			struct umapi_list_head *pos, *n;
			umapi_list_for_each_safe(pos, n, &pHttp->datablocks)
			{
				pBlock = umapi_list_entry(pos,S_BLOCK,head);
				umapi_list_del(&pBlock->head);
				break;
			}
			umapi_os_mutex_unlock(pHttp->mutexHandle);
			if(pBlock && pHttp->flag)
			{
				AVPacket pkt;
				pkt.data = &pBlock->data[6];
				pkt.size = pBlock->datasize;
				struct umapi_list_head *pos, *n;
				umapi_list_for_each_safe(pos, n, &pHttp->sinks)
				{
					S_SINK *sink = umapi_list_entry(pos,S_SINK, head);
					if(sink->writePacket)
					{
						sink->writePacket(sink, &pkt);
					}
				}
			}
			if(pBlock)
			{
				umapi_os_mutex_lock(pHttp->mutexHandle, -1);
				umapi_list_add(&pBlock->head, &pHttp->freeblocks);
				pHttp->blockCount--;
				umapi_os_mutex_unlock(pHttp->mutexHandle);
				LOGI("hdTask[%p] Report have block :%d", pHttp, pHttp->blockCount);
			}else
				break;
			pBlock = UM_NULL;
		}
breakLoop:		
		umapi_os_task_sleep(20);
		continue;
	}
	pHttp->stopflag = UM_TRUE;
}
#define MAXCACHEBLOCKSIZE 20
static UM_S32 init(S_BASICCONTEXT *pHttp, S_SOURCE *pSource)
{
	umapi_os_mutex_create(UM_NULL,UM_OS_MUTEX_ATTR_TIMED_NP,&pHttp->mutexHandle);
	umapi_os_sem_create(UM_NULL,0,MAXCACHESIZE,&pHttp->semhandle);
	umapi_list_create(&pHttp->sinks);

	umapi_os_mutex_lock(pHttp->mutexHandle, -1);
	pHttp->cacheflag = 0;
	if(pHttp->cacheflag)
	{
		umapi_list_create(&pHttp->freeblocks);
		umapi_list_create(&pHttp->datablocks);
		pHttp->pblock = umapi_malloc(sizeof(S_BLOCK) * MAXCACHEBLOCKSIZE);
		if(!pHttp->pblock)
		{
			pHttp->cacheflag = 0;
			LOGS("create http cache blocks");
			goto NOCACHE;
		}
		pHttp->blockCount = 0;
		for(UM_S32 i=0; i<MAXCACHEBLOCKSIZE; i++)
		{
			umapi_sprintf(pHttp->pblock[i].data,"%x\r\n",13160);
			umapi_memcpy(&pHttp->pblock[i].data[1316 * 10 + 6], "\r\n", 2);
			pHttp->pblock[i].datasize = 0;
			umapi_list_add(&pHttp->pblock[i].head, &pHttp->freeblocks);
		}
		pHttp->flag = UM_TRUE;
		pHttp->stopflag = UM_FALSE;
		umapi_os_task_create(UM_NULL, handoutTask, pHttp, 8, 0, &pHttp->taskHandle);
		umapi_os_task_detach(pHttp->taskHandle);
	}
	else
	{
NOCACHE:	
		pHttp->flag = UM_TRUE;
		pHttp->stopflag = UM_TRUE;
	}
	umapi_os_mutex_unlock(pHttp->mutexHandle);
	return UM_SUCCESS;
}

static UM_VOID deinit(S_BASICCONTEXT *pHttp)
{
	pHttp->flag = UM_FALSE;
	if(pHttp->cacheflag){
		umapi_os_sem_post(pHttp->semhandle);
		while(!pHttp->stopflag)
		{
			umapi_os_task_sleep(100);
		}
		umapi_free(pHttp->pblock);
	}
	umapi_os_sem_destroy(pHttp->semhandle);
}

BASICSOURCEFUN(S_BASICCONTEXT, init, deinit);

S_SOURCE streamSource = 
{
    .protocol = "stream",
	.describe = "avPacket==>avPacket || avStream==>avStream",
    .open = open,
    .open2 = open2,
    .addSink = addSink,
    .delSink = delSink,
    .start =  start,
    .stop = stop,
    .destroy = destroy,
    .trace = trace,
    .priSize = sizeof(S_BASICCONTEXT),
};

